package net.wicp.tams.common.binlog.dump;

import static com.lmax.disruptor.RingBuffer.createMultiProducer;

import java.lang.management.ManagementFactory;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.management.InstanceAlreadyExistsException;
import javax.management.MBeanRegistrationException;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.NotCompliantMBeanException;
import javax.management.ObjectName;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ArrayUtils;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkProcessor;
import com.lmax.disruptor.util.DaemonThreadFactory;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.Conf;
import net.wicp.tams.common.apiext.CollectionUtil;
import net.wicp.tams.common.apiext.LoggerUtil;
import net.wicp.tams.common.apiext.StringUtil;
import net.wicp.tams.common.apiext.jdbc.MySqlAssit;
import net.wicp.tams.common.binlog.dump.bean.Dump;
import net.wicp.tams.common.binlog.dump.bean.Dump.DumpBuilder;
import net.wicp.tams.common.binlog.dump.bean.DumpEvent;
import net.wicp.tams.common.binlog.dump.handlerConsumer.BaseDataHander;
import net.wicp.tams.common.binlog.dump.handlerConsumer.BusiHander;
import net.wicp.tams.common.binlog.dump.handlerConsumer.Publisher;
import net.wicp.tams.common.binlog.dump.handlerConsumer.SendHander;
import net.wicp.tams.common.binlog.dump.jmx.DumpControl;
import net.wicp.tams.common.binlog.dump.listener.IBusiFilter;
import net.wicp.tams.common.binlog.dump.listener.IBusiSender;
import net.wicp.tams.common.constant.JvmStatus;
import net.wicp.tams.common.constant.dic.YesOrNo;
import net.wicp.tams.common.jdbc.DruidAssit;

/***
 * 消费模式
 * 
 * @author 偏锋书生
 *
 *         2018年5月27日
 */
@Slf4j
public class MainDump {
	private static final int BUFFER_SIZE = 128 * 1; // 1024 3G以上 256 2.8G 128 2G
	
	public static final String globleDatasourceName="_global";

	// private final ExecutorService executor;// = Executors.newFixedThreadPool(8,
	// DaemonThreadFactory.INSTANCE);

	private final EventFactory<DumpEvent> EVENT_FACTORY = new EventFactory<DumpEvent>() {
		public DumpEvent newInstance() {
			return new DumpEvent();
		}
	};

	private final RingBuffer<DumpEvent> ringBuffer = createMultiProducer(EVENT_FACTORY, BUFFER_SIZE,
			new BlockingWaitStrategy());

	Dump[] dumps = null;

	public void dump() throws SQLException, InstantiationException, IllegalAccessException, ClassNotFoundException {
		// dump初禁用了
		if (!Conf.getBoolean("common.binlog.alone.dump.global.enable")) {
			return;
		}
		Thread.currentThread().setName("Dump-main");
		Map<String, Map<String, String>> dumpConfs = Conf.getPreGroup("common.binlog.alone.dump.ori");
		if (MapUtils.isEmpty(dumpConfs)) {
			System.err.println("----未配置dump任务，不能启动task----");
			log.error("----未配置dump任务，不能启动task----");
			return;
		}
		dumps = new Dump[dumpConfs.size()];
		Properties newprops = Conf.replacePre("common.binlog.alone.dump.global.pool", "common.jdbc.datasource."+globleDatasourceName);
		Conf.overProp(newprops);
		Connection connection = DruidAssit.getConnection(globleDatasourceName);
		int i = 0;
		for (String dumpId : dumpConfs.keySet()) {
			DumpBuilder dumpBuilder = Dump.builder().id(dumpId);
			Map<String, String> dumpConf = dumpConfs.get(dumpId);
			dumpBuilder.db(dumpConf.get("db")).tb(dumpConf.get("tb"))
					.numDuan(Integer.parseInt(dumpConf.get("numDuan")));
			String[] primarys = MySqlAssit.getPrimary(connection, dumpConf.get("db"), dumpConf.get("tb"));
			dumpBuilder.primarys(primarys);
			dumpBuilder.wheresql(StringUtil.hasNull(dumpConf.get("wheresql")));
			dumpBuilder.startId(StringUtil.hasNull(dumpConf.get("startId")));
			dumpBuilder.numDump(
					StringUtil.isNull(dumpConf.get("numDump")) ? null : Long.parseLong(dumpConf.get("numDump")));

			String[][] cols = MySqlAssit.getCols(connection, dumpConf.get("db"), dumpConf.get("tb"), YesOrNo.yes);
			String needCols = dumpConf.get("needCols");
			if (StringUtil.isNull(needCols)) {
				dumpBuilder.needCols(cols[0]);
				dumpBuilder.needColTypes(Dump.convertColumnType(cols[1]));
			} else {
				String[] configCols = needCols.split(",");
				String[] retEles = CollectionUtil.arrayAnd(String[].class, cols[0], configCols);// cols[0]与configCols不能调换顺序
				String[] retTypes = new String[retEles.length];
				for (int j = 0; j < retEles.length; j++) {
					int index = ArrayUtils.indexOf(cols[0], retEles[j]);
					retTypes[j] = cols[1][index];
				}
				dumpBuilder.needCols(retEles);
				dumpBuilder.needColTypes(Dump.convertColumnType(retTypes));
			}
			if (StringUtil.isNotNull(dumpConf.get("busiFilter"))) {
				IBusiFilter<DumpEvent> newInstance = (IBusiFilter<DumpEvent>) Class.forName(dumpConf.get("busiFilter"))
						.newInstance();
				dumpBuilder.busiFilter(newInstance);
			}
			if (StringUtil.isNotNull(dumpConf.get("busiSender"))) {
				String busiSender = StringUtil.trimSpace(dumpConf.get("busiSender"));
				BusiSenderEnum busiSenderEnum = BusiSenderEnum.valueOf(busiSender);
				if (busiSenderEnum != null) {
					busiSender = busiSenderEnum.getClassName();
				}
				IBusiSender<DumpEvent> newInstance = (IBusiSender<DumpEvent>) Class.forName(busiSender).newInstance();
				newInstance.init(dumpId);
				dumpBuilder.busiSender(newInstance);
			}
			dumpBuilder.metric(new DumpGroup(dumpId));
			dumps[i++] = dumpBuilder.build();
		}
		log.info("----------------------启动jmx-------------------------------------");
		try {
			initMbean();// 启动jxmx
		} catch (Exception e) {
			log.error("启动jmx错误", e);
			LoggerUtil.exit(JvmStatus.s15);
		}
		log.info("----------------------配置metrix-------------------------------------");
		// System.setProperty(TsLogger.ENV_FILE_NAME, "dump_" + dumpId);
		// System.setProperty(TsLogger.ENV_FILE_ROOT, String.format("%s/logs/metrics",
		// System.getenv("DUCKULA_DATA")));
		log.info("----------------------启动Disruptor-------------------------------------");
		try {
			disruptorRun();
		} catch (Exception e) {
			log.error("dump失败,将关机，原因：", e);
			LoggerUtil.exit(JvmStatus.s15);
		}
		addTimer();
	}

	SequenceBarrier baseBarrier;
	SequenceBarrier busiBarrier;

	public static Publisher[] publishers;

	private void disruptorRun() throws SQLException {
		// 如果保证顺序需要单线程
		publishers = new Publisher[dumps.length];
		for (int i = 0; i < publishers.length; i++) {
			publishers[i] = new Publisher(ringBuffer, dumps[i]);
		}
		///////////////////////////////// 取基础数据////////////////////////////////////
		Sequence workSequence = new Sequence(-1);
		SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
		int baseDataNum = Conf.getInt("common.binlog.alone.dump.thread.baseDataNum");
		BaseDataHander[] baseDataHanders = new BaseDataHander[baseDataNum];
		for (int i = 0; i < baseDataHanders.length; i++) {
			baseDataHanders[i] = new BaseDataHander(dumps);
		}
		@SuppressWarnings("unchecked")
		WorkProcessor<DumpEvent>[] baseProcessors = new WorkProcessor[baseDataNum];
		for (int i = 0; i < baseProcessors.length; i++) {
			baseProcessors[i] = new WorkProcessor<DumpEvent>(ringBuffer, sequenceBarrier, baseDataHanders[i],
					new IgnoreExceptionHandler(), workSequence);
		}
		///////////////////////////////////////// 业务处理///////////////////////////////////////////////////////////////////////////////
		baseBarrier = ringBuffer.newBarrier(getSeqAry(baseProcessors));
		Sequence busiSequence = new Sequence(-1);
		int busiNum = Conf.getInt("common.binlog.alone.dump.thread.busiNum");
		BusiHander[] busiHanders = new BusiHander[busiNum];
		for (int i = 0; i < busiHanders.length; i++) {
			busiHanders[i] = new BusiHander(dumps);
		}
		@SuppressWarnings("unchecked")
		WorkProcessor<DumpEvent>[] busiProcessors = new WorkProcessor[busiNum];
		for (int i = 0; i < busiProcessors.length; i++) {
			busiProcessors[i] = new WorkProcessor<DumpEvent>(ringBuffer, baseBarrier, busiHanders[i],
					new IgnoreExceptionHandler(), busiSequence);
		}

		///////////////////////////////////////// 发送处理///////////////////////////////////////////////////////////////////////////////
		busiBarrier = ringBuffer.newBarrier(getSeqAry(busiProcessors));
		Sequence sendSequence = new Sequence(-1);
		// 不需要发送只要一个线程就OK了
		int sendNum = Conf.getInt("common.binlog.alone.dump.thread.sendNum");
		SendHander[] sendHanders = new SendHander[sendNum];
		for (int i = 0; i < sendHanders.length; i++) {
			sendHanders[i] = new SendHander(dumps);
		}
		@SuppressWarnings("unchecked")
		WorkProcessor<DumpEvent>[] sendProcessors = new WorkProcessor[sendNum];
		for (int i = 0; i < sendProcessors.length; i++) {
			sendProcessors[i] = new WorkProcessor<DumpEvent>(ringBuffer, busiBarrier, sendHanders[i],
					new IgnoreExceptionHandler(), sendSequence);
		}

		ringBuffer.addGatingSequences(getSeqAry(sendProcessors));
		ExecutorService executor = Executors.newFixedThreadPool(publishers.length + baseDataNum + busiNum + sendNum,
				DaemonThreadFactory.INSTANCE);
		for (int i = 0; i < publishers.length; i++) {
			executor.submit(publishers[i]);
		}
		for (WorkProcessor<DumpEvent> baseProcessor : baseProcessors) {
			executor.submit(baseProcessor);
		}
		for (WorkProcessor<DumpEvent> busiProcessor : busiProcessors) {
			executor.submit(busiProcessor);
		}
		for (WorkProcessor<DumpEvent> sendProcessor : sendProcessors) {
			executor.submit(sendProcessor);
		}
		addShutdownHook();
		beginTime = System.currentTimeMillis();
	}

	private Sequence[] getSeqAry(WorkProcessor<DumpEvent>[] baseProcessors) {
		Sequence[] seqAry = new Sequence[baseProcessors.length];
		for (int i = 0; i < seqAry.length; i++) {
			seqAry[i] = baseProcessors[i].getSequence();
		}
		return seqAry;
	}

	private void addShutdownHook() {
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				log.info("----------------------执行关闭进程 钩子开始-------------------------------------");
				// DisruptorManager.getInst().stop(); // 为什么hold住？
				updateLastId();
				log.info("----------------------执行关闭进程 钩子完成-------------------------------------");
			}
		});
	}

	private static void initMbean() throws InstanceAlreadyExistsException, MBeanRegistrationException,
			NotCompliantMBeanException, MalformedObjectNameException {
		DumpControl control = new DumpControl();
		MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
		mbs.registerMBean(control, new ObjectName("Commons:name=dumpBean"));
		log.info("----------------------MBean注册成功-------------------------------------");
	}

	private void addTimer() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		// 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
		service.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				updateLastId();
			}
		}, 10, 3, TimeUnit.SECONDS);
	}

	private void updateLastId() {
		long min = this.ringBuffer.getMinimumGatingSequence();
		if (ArrayUtils.isEmpty(dumps)) {
			return;
		}
		long timeUse = System.currentTimeMillis() - beginTime;
		for (Dump dump : dumps) {
			log.info("-------------------------table:{}.{}, speed:{},lastId:{}", dump.getDb(), dump.getTb(),
					dump.getMetric().counter_send_num.getCount() * 1000 / timeUse, dump.getLastId());
		}
		// 总体日志
		log.info(
				"-------------------------time:{} minute,undo size:{},sendDuanNo:{},busiDuanNo:{},baseDateDuanNo:{},publisher:{}",
				timeUse / (1000 * 60), ringBuffer.getCursor() - min, min, busiBarrier.getCursor(),
				baseBarrier.getCursor(), ringBuffer.getCursor());
	}

	private static long beginTime;

	/*
	 * public static void main(String[] args) throws IOException, SQLException,
	 * InstantiationException, IllegalAccessException, ClassNotFoundException {
	 * MainDump main = new MainDump(); main.dump(); System.in.read(); }
	 */
}
